package com.lxl.testHd.mypartition;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @author ：e_lixilin
 * @date ：2022/2/25 8:17
 * @description：
 * @modified By：更具车辆类型进行分区,求各个车型的平均值
 */
public class PartDriver {
    static class CarDriveMapper extends Mapper<LongWritable, Text, Text, AvgCarInfo> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            // 输出 :  key 车型, value 行驶信息
            //将一行内容转成string
            String line = value.toString();
            //得到车辆行驶的各个信息（入站时间、出站时间、行驶公里数...）
            String[] fields = line.split("\t");
            //获取车的类别
            String carType = fields[3];
            //获取行驶公里数
            long distance = Long.parseLong(fields[2]);

            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd.HH:mm:ss") ;
            //进站时间
            String startDateStr = fields[0] ;
            //出站时间
            String endDateStr = fields[1] ;
            Date startDate = null ;
            Date endDate = null ;
            try {
                startDate = sdf.parse(startDateStr ) ;
                endDate = sdf.parse(endDateStr ) ;
            }catch(Exception e){
                e.printStackTrace();
            }
            //计算行驶时间（单位：分钟）
            long betweenMinutes = (int)((endDate.getTime() - startDate.getTime())/1000/60);
            /*
             * 输出。 key:车型      value：(行驶距离,行驶时间)
             * 例如：
             *      中型       (300,200)
             *   	小型       (200,105)
             *   	特大型   (100,65)
             */
            context.write(new Text(carType), new AvgCarInfo(distance,betweenMinutes));
        }
    }
    static class CarDriveReduce extends Reducer<Text, AvgCarInfo, Text, AvgCarInfo> {
        @Override
        protected void reduce(Text key, Iterable<AvgCarInfo> values, Context context) throws IOException, InterruptedException {
            //统计车辆个数
            int carCount = 0 ;
            //总行驶公里数
            long sumDistance = 0;
            //总行驶时间
            long sumMinutes = 0 ;
            for (AvgCarInfo carInfo : values) {
                carCount++;
                sumDistance+=carInfo.getDistance();
                sumMinutes += carInfo.getMinutes();
            }
            //平均行驶公里数
            long avgDistance = sumDistance/carCount ;//忽略小数
            //平均行驶时间
            long avgMinutes = sumMinutes/carCount ;
            //平均行驶速度（单位：公里/小时）
            double avgSpeed = avgDistance/(avgMinutes/60.0) ;
            context.write(key, new AvgCarInfo(avgDistance,avgMinutes,avgSpeed));
        }
    }

    public static void main(String[] args) throws Exception {
        args = new String[]{"D:\\big-data\\etl\\mypartition", "D:\\big-data\\etl\\mypartition\\output"};
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass(PartDriver.class);
        job.setMapperClass(CarDriveMapper.class);
        job.setReducerClass(CarDriveReduce.class);
//        //设置自定义分区
//        job.setPartitionerClass(SelfPartitioner.class);
//        //设置ReduceTask的个数，用于指定分区的数量
//        job.setNumReduceTasks(4);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(AvgCarInfo.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(AvgCarInfo.class);
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        boolean res = job.waitForCompletion(true);
        System.exit(res?0:1);
    }
}
